package com.easy.stock.spider.html;

import java.net.InetSocketAddress;
import java.net.URI;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.http.DefaultHttpRequest;
import org.jboss.netty.handler.codec.http.HttpHeaders;
import org.jboss.netty.handler.codec.http.HttpMethod;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpVersion;
import org.jboss.netty.util.HashedWheelTimer;
import org.jboss.netty.util.Timer;
import com.easy.stock.spider.HttpClientPipelineFactory;
import com.easy.stock.spider.Spider;
import com.easy.stock.spider.SpiderTask;
import com.easy.stock.util.Constants;

public abstract class BaseHtmlSpider implements Spider {
	
	protected static Log LOG = LogFactory.getLog("spider");

	protected int connectTimeout=2000;
	protected int requestTimeout=2000;
	protected int taskTimeout=5*60;		// 任务超时时间，单位为秒
	protected int maxParallelTaskNum=5; // 默认值
	protected int maxRetryNum=5;	// 默认值
	protected int queueSize=100000;	// 默认值
	protected List<SpiderTask> taskQueue = new LinkedList<SpiderTask>();
	
	public AtomicInteger parallelNum = new  AtomicInteger(0);

	
	/**
	 * 所有的子spider通过这个接口来往队列里添加抓取任务
	 * 
	 * @param task
	 */
	public void addTask(SpiderTask task) {
		synchronized(taskQueue) {
			if(taskQueue.size()<queueSize)
				taskQueue.add(task);
		}
	}
	
	protected abstract List<SpiderTask> prepareSeeds();
	
	public void setup(Properties props) {
		if(props.containsKey("maxParallelTaskNum"))
			maxParallelTaskNum = Integer.parseInt(props.getProperty("maxParallelTaskNum"));
		if(props.containsKey("maxRetryNum"))
			maxRetryNum = Integer.parseInt(props.getProperty("maxRetryNum"));
		if(props.containsKey("queueSize"))
			queueSize = Integer.parseInt(props.getProperty("queueSize"));
	}
	
	
	public int crawl() throws Exception {
		
		// step1, 准备初始种子
		List<SpiderTask> seeds = prepareSeeds();
		if(seeds==null || seeds.isEmpty()) {
			LOG.info("no seed tasks!");
			return Constants.SUCCESS;
		}
		
		taskQueue.addAll(seeds);
		LOG.info("begin crawl for "+this.getClass().getSimpleName()+" , seed num="+seeds.size());

		// 防止内存泄露
		seeds.clear();
		seeds=null;
		
		// step2, 初始化netty
        ClientBootstrap bootstrap = new ClientBootstrap(
                new NioClientSocketChannelFactory(
                        Executors.newCachedThreadPool(),
                        Executors.newCachedThreadPool()));
		Timer timer = new HashedWheelTimer(); 
        bootstrap.setPipelineFactory(new HttpClientPipelineFactory(timer, requestTimeout));
		bootstrap.setOption("child.connectTimeoutMillis", connectTimeout);
		Random random = new Random(System.currentTimeMillis());
		// step3, 循环处理抓取任务，直到所有任务结束
		while(!taskQueue.isEmpty()) {
			
			Iterator<SpiderTask> iter = taskQueue.iterator();
			while(iter.hasNext()) {
				SpiderTask task = iter.next();
			
				switch(task.getStatus()) {
				
				case SUCCESS:
					// 删除已经成功的任务
					iter.remove();
					parallelNum.decrementAndGet();
					task.getChannel().close();
					LOG.info("success to crawl url : "+task.getUrl());
					break;
					
				case FAILED:
					{
						if(task.getRetryNum()>maxRetryNum) {
							// 删除已经多次失败的任务
							iter.remove();
							parallelNum.decrementAndGet();
							task.getChannel().close();
							LOG.error("fail to crawl url : "+task.getUrl());
							break;
						} else {
							// 重试失败的任务
							task.setRetryNum(task.getRetryNum()+1);
							task.setStatus(SpiderTask.STATUS.READY);
							parallelNum.decrementAndGet();
							task.getChannel().close();
							break;
						}
					}
				
				case READY:
					{
						if(parallelNum.get()>=maxParallelTaskNum) {
							break;
						}
						
						// 启动这个任务
						// 下面的代码，顺序不能调整，防止死锁
						parallelNum.incrementAndGet();
						task.setStatus(SpiderTask.STATUS.RUNNING);
						task.setStartTs((int)(System.currentTimeMillis()/1000));
						if(!prepareGetUrl(bootstrap, task)) {
							task.setStatus(SpiderTask.STATUS.FAILED);
						}					
						break;
					}
					
				case RUNNING: 
					{
//						if(random.nextInt(10000)<=1)
//							LOG.info("task queuesize="+taskQueue.size()+", parallelNum="+parallelNum.get());
						int currentTs = (int)(System.currentTimeMillis()/1000);
						if(currentTs-task.getStartTs()>=taskTimeout) {
							LOG.error("crawl url timeout, url="+task.getUrl());
							task.setStatus(SpiderTask.STATUS.FAILED);
						}
						break;
					}
				default:
					break;
				}
				
			}
			
			// sleep a while for next-round check
			Thread.sleep(100);
		}
		
		LOG.info("finish crawl for "+this.getClass().getSimpleName());
        bootstrap.releaseExternalResources();
        timer.stop();
		// step4, 关闭链接
		return Constants.SUCCESS;
		
	}
	
	private boolean prepareGetUrl(ClientBootstrap bootstrap, SpiderTask task) {
		
		try {
			URI uri = new URI(task.getUrl());
			String host = uri.getHost() == null? "localhost" : uri.getHost();
			int port = uri.getPort();
			if (port == -1) {
			    port = 80;
			}
			
			ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
			future.getChannel().getPipeline().addBefore("timeout", "handler", task.getHandler());
			Channel channel = future.awaitUninterruptibly().getChannel();
			task.setChannel(channel);
	        if (!future.isSuccess()) {
	        	LOG.error("prepareGetUrl connect error, reason=", future.getCause());
	            return false;
	        }
			
			// Prepare the HTTP request.
	        HttpRequest request = new DefaultHttpRequest(
	                HttpVersion.HTTP_1_1, HttpMethod.GET, uri.toASCIIString());
	        request.setHeader(HttpHeaders.Names.HOST, host);
	        request.setHeader(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE);
	        request.setHeader(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP);
	        // Send the HTTP request.
	        channel.write(request);
			return true;
		} catch (Exception e) {
        	LOG.error("prepareGetUrl request error, reason=", e);
			return false;
		}
		
	}
	
	public void setMaxParallelTaskNum(int maxParallelTaskNum) {
		this.maxParallelTaskNum = maxParallelTaskNum;
	}

	public void setMaxRetryNum(int maxRetryNum) {
		this.maxRetryNum = maxRetryNum;
	}

	public void setQueueSize(int queueSize) {
		this.queueSize = queueSize;
	}
	
	public void setConnectTimeout(int connectTimeout) {
		this.connectTimeout = connectTimeout;
	}

	public void setRequestTimeout(int requestTimeout) {
		this.requestTimeout = requestTimeout;
	}

}
